//
// Created by Wayne Wang on 2018/11/8.
//

#ifndef ENGINE_RACE_KEYVALUELOG_H
#define ENGINE_RACE_KEYVALUELOG_H

#include <stdint.h>
#include <string>
#include <sstream>
#include <iostream>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <include/engine.h>
#include <unistd.h>
#include <mutex>
#include <byteswap.h>
#include <vector>
#include <atomic>
#include <algorithm>
#include "include/polar_string.h"
#include "KeyLocation.h"
#include "KeyLogCache.h"
#include "constants.h"

namespace polar_race {

    class KeyValueLog {
    public:

        KeyValueLog(const std::string &path, const int &keyLogId, const int &idInManager, const int valueFd)
                :id(keyLogId), count(0), valueFd(valueFd), initialOffset(idInManager * 256L * 1024 * 1024 * 1024 / KeyLogTotalNum) {
            this->KEY_BUFFER_SIZE = static_cast<size_t>(8 * 1024 * 1024 * 64 / KeyLogTotalNum);
            this->keyMap.reserve(static_cast<unsigned long>(1024 * 1024 * 64 / KeyLogTotalNum));
            pthread_mutex_init(&_mutex, nullptr);

            //获得Key file path
            std::ostringstream ss1;
            ss1 << path << "/key-" << this->id;
            this->keyFilePath = ss1.str();

            //获得value file path

            //获得segment file path
            std::ostringstream ss3;
            ss3 << path << "/segment-" << this->id;
            this->segmentFilePath = ss3.str();

            //打开key文件并且映射mmap
            auto mode =  O_RDWR | O_NOATIME;
            this->keyFd = openOrCreateFile(keyFilePath, KEY_BUFFER_SIZE, mode);
            posix_fadvise(keyFd, 0, KEY_BUFFER_SIZE, POSIX_FADV_SEQUENTIAL);
            this->keyBuffer = mmap(nullptr, KEY_BUFFER_SIZE, PROT_READ | PROT_WRITE,
                                   MAP_SHARED,
                                   this->keyFd, 0);

            //open segment file and mmap it.
            this->segmentFd = openOrCreateFile(segmentFilePath, SegmentSize, mode);
//            posix_fallocate(this->segmentFd, 0, SegmentSize);
            this->segmentBuffer = mmap(nullptr, static_cast<size_t>(SegmentSize),
                                       PROT_READ | PROT_WRITE,
                                       MAP_SHARED,
                                       this->segmentFd, 0);
        }

        ~KeyValueLog() {
            munmap(keyBuffer, KEY_BUFFER_SIZE);
            munmap(segmentBuffer, SegmentSize);
            close(this->keyFd);
            close(this->segmentFd);
            pthread_mutex_destroy(&_mutex);
        }

        void initialize(void * buffer) {

            int maxKeySize = static_cast<int>(KEY_BUFFER_SIZE / 8);
            
            off_t offset = 0;

            int location = 0;
            bool completed = false;
            int keyCount = static_cast<int>(SegmentSize / 8);

            while(!completed) {
                pread64(keyFd, buffer, SegmentSize, offset);
                auto ptr = reinterpret_cast<uint64_t *>(buffer);
                for(int i = 0; i < keyCount; ++i) {
                    if(ptr[i] == 0) {
                        completed = true;
                        break;
                    }
                    keyMap.emplace_back(ptr[i], location);
                    ++location;
                }
                offset += SegmentSize;
            }

            this->currentKeyPosition = location;

            if(!keyMap.empty()) {
                std::sort(keyMap.begin(), keyMap.end(), KeyLocation::compare);
                posix_fadvise(keyFd, 0, KEY_BUFFER_SIZE, POSIX_FADV_DONTNEED);
            }
        }

        bool empty() {
            return keyMap.empty();
        }

        void putValue(const PolarString &key, const PolarString &value, uint64_t key64) {
            pthread_mutex_lock(&_mutex);

            int segmentPosition = currentKeyPosition % SegmentCount;

            memcpy(static_cast<char *>(segmentBuffer) + (segmentPosition * 4096), value.data(), 4096);

            if (segmentPosition == SegmentCount - 1) {
                auto valueFilePosition = (currentKeyPosition - segmentPosition) * ValueSize;
                pwrite64(this->valueFd, segmentBuffer, SegmentSize, valueFilePosition + initialOffset);
            }

            auto *key64Buffer = reinterpret_cast<uint64_t *>(keyBuffer);
            key64Buffer[currentKeyPosition] = key64;

//            fprintf(stderr, "w %lu, %d, %i, %d\n", key64, currentKeyPosition, id, *value.data());

            currentKeyPosition++;

            pthread_mutex_unlock(&_mutex);
        }

        RetCode readValue(const PolarString &key, char *value, uint64_t key64) {
            auto index = searchKey(key64);
            if (unlikely(index == keyMap.end())) {
//                fprintf(stderr, "not found %li, %d, %li, %s\n", k, index, id, key.ToString().data());
                return kNotFound;
            } else {
                auto filePosition = ((long) index->location) * 4096;
                int segmentPosition = currentKeyPosition % SegmentCount;
                long valueFilePosition = ((long) (currentKeyPosition - segmentPosition)) * ValueSize;
                long offset = filePosition - valueFilePosition;
                if (offset >= 0) {
                    memcpy(value, static_cast<char *>(this->segmentBuffer) + offset, 4096);
                } else {
                    pread64(this->valueFd, value, 4096, filePosition + initialOffset);
                }
//                fprintf(stderr, "r %lu, %d, %i, %d\n", key64, index->location, id, *value);
                return kSucc;
            }
        }

        std::vector<KeyLocation>::iterator searchKey(const uint64_t &key) {
            auto upper = findUpper(key);
            if (upper == keyMap.begin() || (--upper)->key != key) {
                return keyMap.end();
            } else {
                return upper;
            }
        }

        std::vector<KeyLocation>::iterator findUpper(const uint64_t &key64) {
            std::vector<KeyLocation>::iterator searched;
            searched = std::upper_bound(keyMap.begin(), keyMap.end(), key64, KeyLocation::uint64CompareKey);
            return searched;
        }

        void normalRange(Visitor &visitor, char * value) {
            char key[8];
            for (auto index = keyMap.begin(); index != keyMap.end(); index++) {
                if (index->key == (index + 1)->key) {
                    continue;
                }
                uint64ToString(index->key, key);
                auto filePosition = ((long) index->location) * 4096;
                int segmentPosition = currentKeyPosition % SegmentCount;
                long valueFilePosition = ((long) (currentKeyPosition - segmentPosition)) * ValueSize;
                long offset = filePosition - valueFilePosition;
                if (offset >= 0) {
                    memcpy(value, static_cast<char *>(this->segmentBuffer) + offset, 4096);
                } else {
                    pread64(this->valueFd, value, 4096, filePosition + initialOffset);
                }
                PolarString k(key, 8);
                PolarString v(value, ValueSize);
                visitor.Visit(k, v);
            }

        }


        void rangeWithCache(Visitor &visitor, KeyLogCache *cache) {
//            fprintf(stderr, "ranging key log %d\n", id);
            char key[8];
            char *value = nullptr;
            int segmentPosition = currentKeyPosition % SegmentCount;
            auto valueFilePosition = ((long) (currentKeyPosition - segmentPosition)) * ValueSize;
            for (auto keyLocation = keyMap.begin(); keyLocation != keyMap.end(); keyLocation++) {
                if (keyLocation->key == (keyLocation + 1)->key) {
                    continue;
                }
                uint64ToString(keyLocation->key, key);
                auto filePosition = keyLocation->location * 4096;
                if (filePosition >= valueFilePosition) {
                    value = static_cast<char *>(this->segmentBuffer) + filePosition - valueFilePosition;
                } else {
                    value = static_cast<char *>(cache->cache) + filePosition;
                }

                PolarString k(key, 8);
                PolarString v(value, ValueSize);
                visitor.Visit(k, v);
            }
        }

        int valueFd;
        int id;
        off_t initialOffset;
        std::vector<KeyLocation> keyMap;

    private:
        int keyFd;
        int segmentFd;
        size_t KEY_BUFFER_SIZE;
        int currentKeyPosition;
        pthread_mutex_t _mutex;
        void *keyBuffer;
        void *segmentBuffer;
        std::atomic_int count;
        std::string keyFilePath;
        std::string segmentFilePath;

    };
}

#endif //ENGINE_RACE_KEYVALUELOG_H
